package com.springboot.gateway.controller;

import com.springboot.gateway.entity.Rocketmq;
import com.springboot.gateway.service.RocketmqService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.util.List;

/**
 * @author anyang
 * Email: 1300378587@qq.com
 * Created Date:2019-11-18 10:32
 */
@RestController
@Slf4j
public class RocketMQController{
    @Value("${server.port}")
    private String port;

    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    @Value("${rocketmq.config.topic}")
    private String mqTopic;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private RocketmqService rocketmqService;

    @GetMapping("/msgList")
    public List<Rocketmq> getMessageList(){
        return  rocketmqService.list();
    }
    
    
    @GetMapping("/mqtest/psend")
    public String psend( String name) throws InterruptedException{
        DefaultMQProducer producer = rocketMQTemplate.getProducer();
        //创建消息，设置topic和body
        Message msg = new Message(mqTopic,"sort消息体".getBytes());
        //发送消息
        Rocketmq rocketmq = sendMessage(producer,msg);

        //保存到数据库
        boolean saveResult = rocketmqService.save(rocketmq);

        log.info("消息发送完毕,消息保存： "+ LocalDateTime.now());
        return " producer服务发送： 端口： " + port+" time: "+ LocalDateTime.now();
    }

    private Rocketmq sendMessage(DefaultMQProducer producer,Message  msg) throws InterruptedException{
        //创建记录保存对象
        String localResult = "";
        Rocketmq rocketmq = new Rocketmq();
        rocketmq.setInfo("sort send message");
        rocketmq.setType("0");
        SendResult result = null ;
        try{
            producer.setProducerGroup(producerGroup);
            //发送消息
            result = producer.send(msg);
            localResult = localResult+result;
            rocketmq.setResult(localResult);
        }catch(RemotingException e){
            rocketmq.setResult(e.getMessage());
            log.error("Remoting Exception ",e);
        }catch(MQClientException e){
            rocketmq.setResult(e.getMessage());
            log.error("MQClient Exception ",e);
        }catch(MQBrokerException e){
            rocketmq.setResult(e.getMessage());
            log.error("MQBroker Exception ",e);
        }
        return  rocketmq;
    }

}
